package com.shujia.flink.sink

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

object Demo1Sink {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val lineDS: DataStream[String] = env.socketTextStream("master", 8888)

    //使用自定义sink
    lineDS.addSink(new MySink)

    env.execute()

  }

}


/**
  * 自定义sink
  *
  * 实现SinkFunction接口，重写里面的invoke
  *
  */

class MySink extends RichSinkFunction[String] {

  /**
    * 在invoke之前执行，在每一个并行度中只执行一次
    *
    */

  override def open(parameters: Configuration): Unit = {
    println("open")
  }

  /**
    *
    * 在invoke之后执行，在每一个并行度中只执行一次
    *
    */
  override def close(): Unit = {
    println("close")
  }

  /**
    * invoke： 每一条数据都会执行一次，在这个方法中将数据发送出去
    *
    * @param value   ： 一行数据
    * @param context ： 上下文对象
    */
  override def invoke(value: String, context: SinkFunction.Context[_]): Unit = {
    println(value)
  }
}
